package pipe

import "sync"

// 生产者，消费者模式消费
// 1. 生产者
// 2. 消费者
// 3. 两个chan，输入chan和输出chan
type InChanV1 chan interface{}
type OutChanV1 chan interface{}
type ProducerFuncV1 func(args ...interface{}) InChanV1
type ConsumerFuncV1 func(in InChanV1) OutChanV1

type CommonFuncV1 func(args ...interface{}) interface{}
type PipeV1 struct {
	Producer ProducerFuncV1
	Consumer ConsumerFuncV1
	Count    int //多少个消费者
}

func NewPipeV1() *PipeV1 {
	return &PipeV1{Count: 1}
}

func (this *PipeV1) SetProducer(p ProducerFuncV1) {
	this.Producer = p
}

func (this *PipeV1) SetConsumer(c ConsumerFuncV1, count int) {
	this.Consumer = c
	this.Count = count
}

func (this *PipeV1) SetProducerNew(p CommonFuncV1) interface{} {
	in := make(InChanV1, 0)
	go func(inFunc CommonFuncV1) {
		defer close(in)
		//需要在函数里面将数据塞进管道
		inFunc(in)
	}(p)
	return in
}

func (this *PipeV1) Excute(args ...interface{}) OutChanV1 {
	//1. 生产
	inC := this.Producer(args)
	out := make(OutChanV1, 0)
	var wg = sync.WaitGroup{}
	//开启消费
	for i := 1; i <= this.Count; i++ {
		wg.Add(1)
		getChan := this.Consumer(inC)
		go func(input OutChanV1) {
			//消费
			defer wg.Done()
			for v := range input {
				out <- v
			}
		}(getChan)
	}
	go func() {
		defer close(out)
		wg.Wait()
	}()
	return out
}
